package org.jetlinks.community.protocol;

import lombok.extern.slf4j.Slf4j;
import org.jetlinks.community.protocol.event.ProtocolChangeEvent;
import org.jetlinks.supports.protocol.management.ProtocolSupportDefinition;
import org.jetlinks.supports.protocol.management.ProtocolSupportManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.util.CollectionUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class DefaultProtocolSupportManager implements ProtocolSupportManager {

    private final ApplicationEventPublisher publisher;

    private ProtocolSupportInfoLoader loader;

    private final Map<String, ProtocolSupportDefinition> cache = new ConcurrentHashMap<>();

    public DefaultProtocolSupportManager(ApplicationEventPublisher publisher) {
        this.publisher = publisher;
    }

    @Autowired
    public void setLoader(ProtocolSupportInfoLoader loader) {
        this.loader = loader;
    }

    @Override
    public Mono<Boolean> store(Flux<ProtocolSupportDefinition> all) {
        return all.collect(Collectors.toMap(ProtocolSupportDefinition::getId, Function.identity()))
                  .flatMap(map -> {
                      cache.putAll(map);
                      return Mono.just(true);
                  });
    }

    @Override
    public Flux<ProtocolSupportDefinition> loadAll() {
        if (CollectionUtils.isEmpty(cache)) {
            return store(loader.loadAll()).flatMapMany(res -> Flux.fromIterable(cache.values()));
        }
        return Flux.fromIterable(cache.values());
    }

    @Override
    public Mono<Boolean> save(ProtocolSupportDefinition definition) {
        cache.put(definition.getId(), definition);
        return ProtocolChangeEvent.of(definition).publish(publisher).thenReturn(true);
    }

    @Override
    public Mono<Boolean> remove(String id) {
        ProtocolSupportDefinition definition = cache.get(id);
        definition.setState((byte) -1);
        cache.remove(id);
        return ProtocolChangeEvent.of(definition).publish(publisher).thenReturn(true);
    }
}
